insert
    overwrite
    table jms_dm.dm_tab_interceptor_waybill_kafka_hi
    partition
    (dt)
select waybill_no
     , input_time
     , date_time
     , status
     , from_unixtime(unix_timestamp(), 'yyyy-MM-dd HH:mm:ss') as sys_time --系统运行时间
     , '{{ execution_date | cst_hour }}'                      as dt
from (
         select t1.waybill_no
              , t1.status
              , t1.input_time
              , t1.date_time
              , case
                    when t1.waybill_no = t2.waybill_no and t1.status <> t2.status then 1
                    when t2.waybill_no is null and t1.status = 1 then 2
                    else 0 end as falg

         from (
                  select waybill_no
                       , status
                       , input_time
                       , date_time
                  from jms_dm.dm_tab_interceptor_waybill_kafka_middle_hi
                  where dt = '{{ execution_date | cst_hour }}'
                    and waybill_no is not null
              ) t1
                  left join
              (
                  select waybill_no
                       , status
                  from jms_dm.dm_tab_interceptor_waybill_kafka_middle_hi
                  where dt = '{{ execution_date | hour_add(-1) | cst_hour }}'
              ) t2
              on t1.waybill_no = t2.waybill_no
     ) a
where falg <> 0
;
